package top.houry.springbootkafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class ProducerTest {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();

        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "182.61.19.148:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());


       // properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "boot-kafka");
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        //producer.initTransactions();
        producer.send(new ProducerRecord<>("quickstart-events", "添加的信息"), (recordMetadata, e) -> {
            System.out.println("ddddd" + recordMetadata.toString());
            System.out.println("e  " + e);
        });
//        producer.send(new ProducerRecord<>("quickstart-events", "firstValue"));
        System.out.println("====");

        while (true) {
            TimeUnit.HOURS.sleep(1);
        }

//        producer.initTransactions();
//        producer.beginTransaction();
//        try {
//            producer.send(new ProducerRecord<>("first", "firstValue"), (recordMetadata, e) -> System.out.println(recordMetadata.toString()));
//        } catch (Exception e) {
//            producer.abortTransaction();
//        } finally {
//            producer.close();
//        }

    }
}
